package com.tree.zk.autoconfig;

import com.alibaba.fastjson.JSON;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.List;

/**
 * package com.tree.zk
 * description:
 *
 * 参考：https://blog.csdn.net/cowbin2012/article/details/85337250
 *
 * @author tree
 * @date 2019-08-22 09:59
 */
@Data
@Slf4j
public class ZkClient {

    private CuratorFramework curatorFramework;

    public TreeCache cache;

    private ZookeeperProperties zookeeperProperties;

    public ZkClient(ZookeeperProperties zookeeperProperties) {
        this.zookeeperProperties = zookeeperProperties;
    }

    public void init(){
        try {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(
                    zookeeperProperties.getBaseSleepTimeMs(),zookeeperProperties.getMaxRetries());
            CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                    .connectString(zookeeperProperties.getServerUrl())
                    .retryPolicy(retryPolicy)
                    .sessionTimeoutMs(zookeeperProperties.getSessionTimeoutMs())
                    .connectionTimeoutMs(zookeeperProperties.getConnectionTimeoutMs())
                    .namespace(zookeeperProperties.getNamespace());
            if(!StringUtils.isEmpty(zookeeperProperties.getDigest())){
                builder.authorization("digest",zookeeperProperties.getDigest().getBytes());
                builder.aclProvider(new ACLProvider() {
                    @Override
                    public List<ACL> getDefaultAcl() {
                        return ZooDefs.Ids.CREATOR_ALL_ACL;
                    }

                    @Override
                    public List<ACL> getAclForPath(String path) {
                        return ZooDefs.Ids.CREATOR_ALL_ACL;
                    }
                });

            }
            curatorFramework = builder.build();
            curatorFramework.start();

            initLocalCache("/");
            curatorFramework.getConnectionStateListenable().addListener(new ConnectionStateListener() {
                @Override
                public void stateChanged(CuratorFramework client, ConnectionState newState) {
                    switch (newState){
                        case LOST:
                            //连接丢失
                            log.info("lost session with zookeeper");
                            break;
                        case RECONNECTED:
                            //连接丢失
                            log.info("RECONNECTED session with zookeeper");
                            break;
                        case CONNECTED:
                            //连接丢失
                            log.info("CONNECTED session with zookeeper");
                            break;
                        case SUSPENDED:
                            //连接丢失
                            log.info("SUSPENDED session with zookeeper");
                            break;
                        default:
                            //连接丢失
                            log.info("no with zookeeper");
                            break;
                    }

                }
            });
        }catch (Exception e){
            log.error("init zk error",e);
        }

    }

    /**
     * 初始化本地缓存
     * @param watchRootPath
     */
    private void initLocalCache(String watchRootPath) throws Exception {
        cache = new TreeCache(curatorFramework,watchRootPath);
        cache.getListenable().addListener(new TreeCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                log.info("event:" + event.getType() +
                        " |path:" + (null != event.getData() ? event.getData().getPath() : null));

                if(event.getData()!=null && event.getData().getData()!=null){
                    log.info("发生变化的节点内容为：" + JSON.toJSONString(event,true));
                }
                /**
                 *
                 * 2019-08-23 11:49:49 [Curator-TreeCache-0] INFO  c.t.z.a.ZkClient - event:NODE_UPDATED |path:/20190822
                 * 2019-08-23 11:49:49 [Curator-TreeCache-0] INFO  c.t.z.a.ZkClient - 发生变化的节点内容为：{
                 * 	"data":{
                 * 		"data":"MDgyMw==",
                 * 		"path":"/20190822",
                 * 		"stat":{
                 * 			"aversion":0,
                 * 			"ctime":1566460024536,
                 * 			"cversion":0,
                 * 			"czxid":35,
                 * 			"dataLength":4,
                 * 			"ephemeralOwner":0,
                 * 			"mtime":1566532189543,
                 * 			"mzxid":243,
                 * 			"numChildren":0,
                 * 			"pzxid":35,
                 * 			"version":11
                 *                }* 	},
                 * 	"type":"NODE_UPDATED"
                 * }
                 */

            }
        });
        cache.start();
    }


    public void destroy(){
        curatorFramework.close();

    }

    /**
     * 创建节点
     * @param mode       节点类型
     *                   1、PERSISTENT 持久化目录节点，存储的数据不会丢失。
     *                   2、PERSISTENT_SEQUENTIAL顺序自动编号的持久化目录节点，存储的数据不会丢失
     *                   3、EPHEMERAL临时目录节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除
     *                   4、EPHEMERAL_SEQUENTIAL临时自动编号节点，一旦创建这个节点的客户端与服务器端口也就是session 超时，这种节点会被自动删除，并且根据当前已近存在的节点数自动加 1，然后返回给客户端已经成功创建的目录节点名。
     * @param path  节点名称
     * @param nodeData  节点数据
     */
    public void createNode(CreateMode mode, String path , String nodeData){
        try {
            curatorFramework.create().creatingParentsIfNeeded().withMode(mode).forPath(path,nodeData.getBytes());
        } catch (Exception e) {
            e.printStackTrace();
            log.error("create node error",e);
        }
    }

    public void createNode(CreateMode mode, String path){

        try {
            curatorFramework.create().creatingParentsIfNeeded().withMode(mode).forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("create node error",e);
        }
    }

    /**
     * 删除节点数据
     * @param path
     */
    public void deleteNode(final String path){
        deleteNode(path,true);
    }

    /**
     * 删除节点数据
     * @param path
     * @param deleteChildren   是否删除子节点
     */

    private void deleteNode(String path, boolean deleteChildren) {
        try {
            if (deleteChildren) {
                //guaranteed()删除一个节点，强制保证删除,
                // 只要客户端会话有效，那么Curator会在后台持续进行删除操作，直到删除节点成功
                curatorFramework.delete().guaranteed().deletingChildrenIfNeeded().forPath(path);
            }else{
                curatorFramework.delete().guaranteed().forPath(path);
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("delete node error",e);
        }
    }

    /**
     * 设置指定节点的数据
     * @param path
     * @param nodeData
     */
    public void setNodeData(String path, String nodeData){
        try {
            curatorFramework.setData().forPath(path,nodeData.getBytes());
        } catch (Exception e) {
            log.error("set node data error",e);
            e.printStackTrace();
        }
    }

    /**
     * 获取指定节点的数据
     * @param path
     * @return
     */
    public String getNodeData(String path) {
        try {
            return  new String (curatorFramework.getData().watched().forPath(path),"UTF-8");
        } catch (Exception e) {
            log.error("get node data error",e);
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 获取数据时先同步
     * @param path
     * @return
     */
    public String synNodeData(String path){
        curatorFramework.sync();
        return getNodeData( path);
    }


    /**
     * 判断路径是否存在
     *
     * @param path
     * @return
     */
    public boolean existPath(String path){
        curatorFramework.sync();
        try {
            return null != curatorFramework.checkExists().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 获取节点的子节点
     * @param path
     * @return
     */
    public List<String> getChildren(String path){
        List<String> childrenList = new ArrayList<>();
        try {
            childrenList = curatorFramework.getChildren().forPath(path);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return childrenList;
    }

//    public Object getShareReentrantLock(String lockPath, long time, SRLockDealCallback<?> dealCallback){
//        InterProcessMutex interProcessMutex = new InterProcessMutex(curatorFramework,)
//    }

    public static void main(String[] args) {

    }
}
